package filter;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Collections;

/**
 * 2 * @Author: 王杰
 * 3 * @Date: 2020/11/18 12:52
 * 4
 */
public class Demo02 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启checkpoint
        env.enableCheckpointing(5000);

        //设置并行度为1
        env.setParallelism(1);

        //将快照保存到本地文件或者hdfs
        //env.setStateBackend(new FsStateBackend("hdfs://node1:8020/flink-checkpoints/"));

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        /**
         * 设置重启策略，默认的重启策略是不停的重启
         * 固定延迟重启策略: 程序出现异常的时候，重启3次，每次延迟5秒钟重启，超过3次，程序退出
         */
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));

        DataStreamSource<String> input = env.readTextFile("data/ip31SampleEventof1_Schema.csv");


        SingleOutputStreamOperator<String> result = input.filter(new RichFilterFunction<String>() {

            private ListState<Long> fisrtline = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ListStateDescriptor<Long> descriptor = new ListStateDescriptor<Long>(
                        "offsetState",
                        TypeInformation.of(new TypeHint<Long>() {
                        })
                );
                fisrtline = getRuntimeContext().getListState(descriptor);
            }

            @Override
            public boolean filter(String value) throws Exception {
                if (fisrtline == null) {
                    fisrtline.update(Collections.singletonList(1L));
                    return true;
                }
                if (value.contains("SampleTime")) {
                    return false;
                }
                return true;
            }
        });

        result.print();
        env.execute();
    }
}
